package com.atguigu.pro2

import java.net.URL
import java.util

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * @description: 优化后的登录失败检测,2s后再报警存在漏洞
 * @time: 2021/4/6 10:29
 * @author: baojinlong
 * */


object LoginFail2 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/LoginLog.csv")
    //val inputStream: DataStream[String] = environment.readTextFile(resource.getPath)
    val inputStream: DataStream[String] = environment.readTextFile("src/main/resources/LoginLog.csv")
    // 转换成样例类行,并提起时间戳和watermark

    val loginEventStream: DataStream[LoginEvent] = inputStream
      .map(data => {
        val arr: Array[String] = data.split(",")
        LoginEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      // 3秒延时
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
        override def extractTimestamp(t: LoginEvent): Long = t.timestamp * 1000
      })

    // 进行判断和检测,如果2秒之内连续登录失败,输出报警信息
    val loginFailWarningStream: DataStream[LoginFailWarning] = loginEventStream
      .keyBy(_.userId)
      .process(new LoginFailWarningResult2)
    // loginFailWarningStream> LoginFailWarning(1035,1558430842,1558430843,login fail 2 times in 2s)
    // loginFailWarningStream> LoginFailWarning(1035,1558430843,1558430844,login fail 2 times in 2s)
    loginFailWarningStream.print("loginFailWarningStream")
    environment.execute("loginFailWarningStreamTest")
  }
}


// 连续登录failTimes就报警
class LoginFailWarningResult2 extends KeyedProcessFunction[Long, LoginEvent, LoginFailWarning] {
  // 定义状态,保存当前所有的登录失败事件,保存定时器的时间戳
  lazy val loginFailListState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("loginFailListState", classOf[LoginEvent]))

  override def processElement(loginEvent: LoginEvent, context: KeyedProcessFunction[Long, LoginEvent, LoginFailWarning]#Context, collector: Collector[LoginFailWarning]): Unit = {
    // 判断当前登录事件是成功还是失败
    if ("fail".equals(loginEvent.eventType)) {
      // 如果是失败,进一步判断
      val iterator: util.Iterator[LoginEvent] = loginFailListState.get.iterator
      if (iterator.hasNext) {
        // 如果有数据那么判断两次失败的时间差
        val firstFailEvent: LoginEvent = iterator.next
        if (loginEvent.timestamp < firstFailEvent.timestamp + 2) {
          // 如果在2s之内直接输出报警信息
          collector.collect(LoginFailWarning(loginEvent.userId, firstFailEvent.timestamp, loginEvent.timestamp, "login fail 2 times in 2s"))
        }
        // 不管是否报警,当前都已经处理完毕,将状态更新为最近一次登录失败的事件
        loginFailListState.clear()
        loginFailListState.add(loginEvent)
      } else {
        // 如果没有直接把当前事件添加到集合中
        loginFailListState.add(loginEvent)
      }
    } else {
      // 如果是成功那么直接定时器和清空状态
      loginFailListState.clear()
    }
  }

}
